package com.allenkerr.rpc.server;

import com.allenkerr.rpc.common.RpcDecoder;
import com.allenkerr.rpc.common.RpcEncoder;
import com.allenkerr.rpc.common.RpcRequest;
import com.allenkerr.rpc.common.RpcResponse;
import com.allenkerr.rpc.registry.ServiceRegistry;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.apache.commons.collections4.MapUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.util.HashMap;
import java.util.Map;

/**
 * 框架的RPC服务器，用于将用户系统的业务类发布为RPC服务。
 * 使用时可由用户通过配置spring-bean的方式注入到用户的业务系统中。
 * 由于本类实现了ApplicationContextAware和InitializingBean接口，spring构造本对象时会
 * 调用setApplicationContext()方法，从而可以在方法中通过自定义注解获得用户的业务接口和实现类，
 * 还会调用afterPropertiesSet()方法，在方法中启动netty服务器。
 * 1. spring初始化类，获取被注解的类
 * 2. 启动netty server，接收client发送的消息
 * 3. 注册zookeeper
 * <p>
 * Created by 小柯(Allen Kerr) on 17.5.23 19:29.
 */
public class RpcServer implements ApplicationContextAware, InitializingBean {

    private static final Logger LOGGER = LoggerFactory.getLogger(RpcServer.class);

    /**
     * rpc-server服务器绑定的地址和端口
     */
    private String serverAddress;

    /**
     * 服务注册
     */
    private ServiceRegistry serviceRegistry;

    //用于存储业务接口和实现类的实例对象(由spring创建)
    private Map<String, Object> handlerMap = new HashMap<>();

    /*public RpcServer(String serverAddress) {
        this.serverAddress = serverAddress;
    }*/

    /**
     * 服务器绑定的地址和端口，由spring在实例化本类时从配置文件读入
     *
     * @param serverAddress
     * @param serviceRegistry
     */
    public RpcServer(String serverAddress, ServiceRegistry serviceRegistry) {
        this.serverAddress = serverAddress;
        //向zookeeper注册名称服务的工具类
        this.serviceRegistry = serviceRegistry;
    }

    /**
     * 通过注解，获取标注了 @RpcService 注解的业务类（接口及impl对象），
     * 并将它放到handlerMap中
     *
     * @param ctx
     * @throws BeansException
     */
    @Override
    public void setApplicationContext(ApplicationContext ctx) throws BeansException {
        Map<String, Object> serviceBeanMap = ctx.getBeansWithAnnotation(RpcService.class);
        if (MapUtils.isNotEmpty(serviceBeanMap)) {
            for (Object serviceBean : serviceBeanMap.values()) {
                //从业务类的自定义注解中获取value，从而获取到业务接口的全名
                String interfaceName = serviceBean.getClass().getAnnotation(RpcService.class).value().getName();
                handlerMap.put(interfaceName, serviceBean);
            }
        }
    }

    /**
     * 在此启动netty服务，绑定handler流水线：
     * 1、接收请求数据进行反序列化得到request对象
     * 2、根据request中的参数，让RpcHandler从handlerMap中找到与接口名对应的实现类，调用指定方法，获取返回结果
     * 3、将业务调用结果封装到response中并序列化后返回给客户端
     *
     * @throws Exception
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        //EventLoopGroup可以理解为是一个线程池，这个线程池用来处理连接、接受数据、发送数据
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            //引导类
            ServerBootstrap bootstrap = new ServerBootstrap();

            //装配bootstrap
            bootstrap.group(bossGroup, workerGroup)//多线程处理
                    .channel(NioServerSocketChannel.class)//指定通道类型为NioServerSocketChannel，一种异步模式
                    .childHandler(new ChannelInitializer<SocketChannel>() {//业务处理类
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast(new RpcDecoder(RpcRequest.class))//注册解码handler，In 1
                                    .addLast(new RpcEncoder(RpcResponse.class))//注册编码handler，Out（out不能放在最后！）
                                    .addLast(new RpcHandler(handlerMap));//注册rpcHandler，In 2
                        }
                    }).option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true);

            String[] array = serverAddress.split(":");
            String host = array[0];
            int port = Integer.parseInt(array[1]);

            //绑定服务器，调用sync()方法会阻塞直到服务器完成绑定
            ChannelFuture future = bootstrap.bind(host, port).sync();
            LOGGER.info("Server started, On {}:{}", host, port);
            if (serviceRegistry != null) {
                serviceRegistry.register(serverAddress);
            }
            //然后服务器等待通道关闭，因为使用sync()，所以关闭操作也会被阻塞。
            future.channel().closeFuture().sync();
        } finally {
            //关闭线程池
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
            //所有工作完成，关闭线程池
            /*workerGroup.shutdownGracefully().sync();
            bossGroup.shutdownGracefully().sync();*/
        }

    }
}
